import sys
import asyncio
from threading import Thread
from PyQt5.QtCore import QThread, pyqtSignal
from MyTT import EMA
import numpy as np
import pandas as pd
from tqsdk import TqApi
from db_manager import connect_db, get_monitored_products
import time 
from tqsdk import TqApi, TqAuth, TqAccount, TqSim,TqKq

class AsyncOpenCloseStrategy(QThread):
    finished_signal = pyqtSignal()
    error_signal = pyqtSignal(str)

    def __init__(self, symbols, lots):
        super().__init__()
        self.symbols = symbols
        self.lots = lots
        self.running = False
        self.active_symbols = set()
        self.api = None
        self.loop = None  # 添加事件循环引用

    def run(self):
        """启动策略"""
        self.running = True
        try:
            # 创建并设置事件循环
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
            
            # 初始化API（必须在事件循环设置后）
            tq_kq = TqKq()
            self.api = TqApi(account=tq_kq, auth=TqAuth("cps168", "alibaba999"), loop=self.loop)
            
            # 运行策略主循环
            self.loop.run_until_complete(self._run_all_symbols())
            
        except Exception as e:
            self.error_signal.emit(f"策略运行出错: {e}")
        finally:
            pass
            # 确保事件循环已关闭
            # if self.loop and not self.loop.is_closed():
            #     self.loop.run_until_complete(self.loop.shutdown_asyncgens())
            #     self.loop.close()
            # 清理资源
            self.running = False
            if self.api:
                self.api.close()
            if self.loop:
                self.loop.close()
            self.finished_signal.emit()

    async def _run_all_symbols(self):
        """并行处理所有品种"""
        print(f"开始处理品种列表: {self.symbols}")
        try:
            tasks = []
            for symbol in self.symbols:
                task = self.loop.create_task(self._process_product(symbol))
                tasks.append(task)
            
            # 添加超时机制
            #await asyncio.wait_for(asyncio.gather(*tasks), timeout=None)
            await asyncio.gather(*tasks)
            
        except asyncio.TimeoutError:
            print("策略执行超时")
        except Exception as e:
            print(f"策略执行出错: {e}")



    async def _process_product(self, symbol):
        """单个品种处理协程"""
        self.active_symbols.add(symbol)
        try:  
            quote = await self.api.get_quote(symbol)  # 添加await
            klines = await self.api.get_kline_serial(symbol, 1*60, 500)  # 添加await
            
            while self.running:
                # 等待行情更新
                await self.api.wait_update(deadline=time.time() + 1)
                
                if self.api.is_changing(quote):
                    print(f"{time.strftime('%X')} {symbol} 行情已更新，当前价格: {quote.last_price}")
                    q = (3*klines.close + klines.low + klines.open + klines.high) / 6
                    ma_values = self._calculate_ma(q)
                    short_ma = ma_values['ma']
                    ema_7 = ma_values['ema']
                    
                    position = await self.api.get_position(symbol)  # 添加await
                    
                    # 平空仓逻辑
                    if short_ma[-2] > ema_7[-2]:  
                        if position.pos_short > 0:
                            self.api.insert_order(symbol, direction="BUY", offset="CLOSE", 
                                            volume=position.pos_short)
                    # 开多仓逻辑
                    if position.pos_long == 0 and position.pos_short == 0 and short_ma[-3] < ema_7[-3] and short_ma[-2] > ema_7[-2]:
                        self.api.insert_order(symbol, direction="BUY", offset="OPEN", 
                                        volume=self.lots)
                        
                    # 平多仓逻辑
                    if short_ma[-2] < ema_7[-2]:  
                        if position.pos_long > 0:
                            self.api.insert_order(symbol, direction="SELL", offset="CLOSE", 
                                            volume=position.pos_long)
                    # 开空仓逻辑
                    if position.pos_long == 0 and position.pos_short == 0 and short_ma[-3] > ema_7[-3] and short_ma[-2] < ema_7[-2]:
                        self.api.insert_order(symbol, direction="SELL", offset="OPEN", 
                                        volume=self.lots)
                #print("************************策略运行结束************************")
                # deadline = time.time() + 1
                # while asyncio.run(self.api.wait_update(deadline=deadline)):  # 等待1秒
                #     pass

        except Exception as e:
            print(f"Error processing symbol {symbol}: {e}")                        
        finally:
            self.active_symbols.remove(symbol)



    # def _calculate_ma(self, q):
    #     """自定义权重计算（与原版完全一致）"""
    #     weights = np.arange(26, 0, -1)
    #     ma = []
    #     q_array = np.array(q)
    #     for i in range(len(q_array)):
    #         if i < 26:
    #             ma.append(np.nan)
    #             continue
    #         window = q_array[i-26:i+1]
    #         ma.append(np.dot(window, weights) / 351)
        
    #     ema_7 = EMA(pd.Series(ma), 7)
    #     return {'ma': ma, 'ema': ema_7.tolist()}

    def _calculate_ma(self, q):
        """自定义权重计算（与原版完全一致）"""
        weights = np.arange(26, 0, -1)
        ma = []
        q_array = np.array(q)
        for i in range(len(q_array)):
            if i < 25:  # 修改为25，因为窗口大小现在是26
                ma.append(np.nan)
                continue
            window = q_array[i-25:i+1]  # 修改为i-25，这样窗口大小就是26
            ma.append(np.dot(window, weights) / 351)
        
        ema_7 = EMA(pd.Series(ma), 7)
        return {'ma': ma, 'ema': ema_7.tolist()}

    def stop(self):
        """停止策略"""
        self.running = False
        if hasattr(self, 'loop') and self.loop:
            # 安全停止事件循环
            self.loop.call_soon_threadsafe(self.loop.stop)
            self.loop.close()